package org.mq.mymq.broker.event.listener;

import com.alibaba.fastjson.JSON;
import org.mq.mymq.broker.cache.CommonCache;
import org.mq.mymq.broker.event.model.PushMsgEvent;
import org.mq.mymq.common.cache.BrokerServerSyncFutureManager;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.MessageDTO;
import org.mq.mymq.common.dto.SendMessageToBrokerResponseDTO;
import org.mq.mymq.common.dto.SlaveSyncRespDTO;
import org.mq.mymq.common.enums.*;
import org.mq.mymq.common.event.Listener;
import org.mq.mymq.common.remote.SyncFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @ClassName PushMsgListener
 * @Description
 * @Author jiarong_ye
 * @Date 2025/6/27 15:48
 * @Version 1.0
 */
public class PushMsgListener implements Listener<PushMsgEvent> {
    private static final Logger log = LoggerFactory.getLogger(PushMsgListener.class);
    @Override
    public void onReceive(PushMsgEvent event) throws Exception {
        // 消息写入commitLog
        // 写入comsumerQueue
        MessageDTO messageDTO = event.getMessageDTO();
        CommonCache.getCommitLogAppendHandler().appendMsg(messageDTO);
        int sendWay = messageDTO.getSendWay();
        boolean isAsyncSend = sendWay == MessageSendWay.ASYNC.getCode();
        boolean isClusterMode = BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(CommonCache.getGlobalProperties().getBrokerClusterMode());
        boolean isMasterNode = "master".equals(CommonCache.getGlobalProperties().getBrokerClusterRole());
        if (isClusterMode) {
            // 集群版本处理
            if (isMasterNode) {
                // master节点处理逻辑
                // 1. 主节点转发同步数据请求给从节点
                CommonCache.getSlaveChannelMap().values().forEach(channel -> {
                    channel.writeAndFlush(new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(), JSON.toJSONBytes(messageDTO)));
                });
                if (isAsyncSend) {
                    return;
                }
                // 判断当前主节点是否还有从节点（master节点发送完数据后，从节点断开，为了保证高可用，这种情况也默认是成功ack）
                if (CommonCache.getSlaveChannelMap().isEmpty()) {
                    SendMessageToBrokerResponseDTO responseDTO = new SendMessageToBrokerResponseDTO();
                    responseDTO.setMsgId(messageDTO.getMsgId());
                    responseDTO.setStatus(SendMessageToBrokerResponseStatus.SUCCESS.getCode());
                    responseDTO.setDesc("send msg success, but current time has no slave node!");
                    TcpMsg respMsg = new TcpMsg(BrokerResponseCode.SEND_MSG_RESP.getCode(), JSON.toJSONBytes(responseDTO));
                    event.getChannelHandlerContext().writeAndFlush(respMsg);
                    return;
                }
                // 正常情况 需要接收从节点返回信息
                SyncFuture syncFuture = new SyncFuture();
                syncFuture.setMsgId(messageDTO.getMsgId());
                BrokerServerSyncFutureManager.put(messageDTO.getMsgId(), syncFuture);
                SyncFuture slaveSyncAckRespFuture = BrokerServerSyncFutureManager.get(messageDTO.getMsgId());
                if (slaveSyncAckRespFuture != null) {
                    SlaveSyncRespDTO slaveSyncRespDTO = null;
                    SendMessageToBrokerResponseDTO responseDTO = new SendMessageToBrokerResponseDTO();
                    responseDTO.setMsgId(messageDTO.getMsgId());
                    responseDTO.setStatus(SendMessageToBrokerResponseStatus.FAIL.getCode());
                    try {
                        //主从网络延迟非常严重
                        slaveSyncRespDTO = (SlaveSyncRespDTO) slaveSyncAckRespFuture.get(3, TimeUnit.SECONDS);
                        if(slaveSyncRespDTO.isSyncSuccess()) {
                            responseDTO.setStatus(SendMessageToBrokerResponseStatus.SUCCESS.getCode());
                        }
                        //超时等同步一系列问题全部注入到响应体中返回给到客户端
                    } catch (InterruptedException e) {
                        responseDTO.setDesc("Slave node sync fail! Sync task had InterruptedException!");
                        log.error("slave sync error is:",e);
                    } catch (ExecutionException e) {
                        responseDTO.setDesc("Slave node sync fail! Sync task had ExecutionException");
                        log.error("slave sync error is:",e);
                    } catch (TimeoutException e) {
                        responseDTO.setDesc("Slave node sync fail! Sync task had TimeoutException");
                        log.error("slave sync error is:",e);
                    } catch (Exception e) {
                        responseDTO.setDesc("Slave node sync unKnow error! Sync task had Exception");
                        log.error("slave sync unKnow error is:",e);
                    }
                    //响应返回给到客户端，完成主从复制链路效果
                    TcpMsg responseMsg = new TcpMsg(BrokerResponseCode.SEND_MSG_RESP.getCode(), JSON.toJSONBytes(responseDTO));
                    event.getChannelHandlerContext().writeAndFlush(responseMsg);                }
            } else {
                // slave节点处理逻辑：接收来自master节点的数据，
                // 如果是同步发送，返回从节点响应信号给master，异步发送直接return
                if (isAsyncSend) {
                    return;
                }
                SlaveSyncRespDTO slaveSyncRespDTO = new SlaveSyncRespDTO();
                slaveSyncRespDTO.setMsgId(messageDTO.getMsgId());
                slaveSyncRespDTO.setSyncSuccess(true);
                event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.SLAVE_SYNC_RESP.getCode(), JSON.toJSONBytes(slaveSyncRespDTO)));
            }
        } else {
            // 单机版本处理：同步发送方式返回响应信号，异步发送直接return
            if (isAsyncSend) {
                return;
            }
            SendMessageToBrokerResponseDTO sendMessageToBrokerResponseDTO = new SendMessageToBrokerResponseDTO();
            sendMessageToBrokerResponseDTO.setStatus(SendMessageToBrokerResponseStatus.SUCCESS.getCode());
            sendMessageToBrokerResponseDTO.setMsgId(messageDTO.getMsgId());
            TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.SEND_MSG_RESP.getCode(), JSON.toJSONBytes(sendMessageToBrokerResponseDTO));
            // 返回ack信号
            event.getChannelHandlerContext().writeAndFlush(tcpMsg);
        }
    }
}
